package com.ferry.lock.engine.zookeeper;

import com.ferry.lock.config.DefaultConfig;
import com.ferry.lock.config.engine.EngineConfig;
import com.ferry.lock.config.engine.ZkConfig;
import com.ferry.lock.engine.AbstractLockEngine;
import com.ferry.lock.engine.zookeeper.menagerie.DefaultZKSessionManager;
import com.ferry.lock.engine.zookeeper.menagerie.ZKSessionManager;
import com.ferry.lock.engine.zookeeper.menagerie.locks.ReentrantZKLock;
import com.ferry.lock.engine.zookeeper.menagerie.locks.ReentrantZKReadWriteLock;
import com.ferry.lock.enumValue.Env;
import com.ferry.lock.lock.IReentrantLock;
import com.ferry.lock.lock.IReentrantReadWriteLock;
import com.ferry.lock.lock.impl.ReentrantReadWriteLock;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created by daiyong
 */
public class ZKLockEngine extends AbstractLockEngine {

	private static final Logger LOGGER = LoggerFactory.getLogger(ZKLockEngine.class);

	private String host;
	private String baseLockPath;
	private int timeout;

	private static final Lock lock = new ReentrantLock();

	private ZKSessionManager zkSessionManager;

	public ZKLockEngine() {
		name = "ZK";
	}

	public boolean init(EngineConfig engineConfig) {
		lock.lock();
		try {
			if (initialized) {
				return true;
			}

			loadConfig(engineConfig);

			zkSessionManager = new DefaultZKSessionManager(host, timeout);
			initialized = true;
			return true;

		} catch (Exception e) {
			LOGGER.error("ZKLockEngine int fail", e);
		} finally {
			lock.unlock();
		}

		return false;
	}

	public void destroy() {
		if (null != zkSessionManager) {
			zkSessionManager.closeSession();
		}
	}

	public IReentrantLock getReentrantLock(String name) {
		checkLockName(name);
		return new ReentrantZKLock(baseLockPath + "/" + name, zkSessionManager);
	}

	public IReentrantLock getReentrantLock(String name, int expireTime) {
		return getReentrantLock(name);
	}

	public IReentrantLock getReentrantLock(String name, int expireTime, int retry) {
		return getReentrantLock(name);
	}

	public IReentrantReadWriteLock getReentrantReadWriteLock(String name) {
		checkLockName(name);
		return new ReentrantZKReadWriteLock(baseLockPath + "/" + name, zkSessionManager);
	}

	public IReentrantReadWriteLock getReentrantReadWriteLock(String name, int expireTime) {
		return getReentrantReadWriteLock(name);
	}

	public IReentrantReadWriteLock getReentrantReadWriteLock(String name, int expireTime, int retry) {
		return getReentrantReadWriteLock(name);
	}

	public Object getProcessor() {
		return zkSessionManager;
	}

	private void checkLockName(String name) {
		if (!StringUtils.isNotBlank(name)) {
			throw new IllegalArgumentException("lockName 不能为空");
		}
	}

	private void loadConfig(EngineConfig engineConfig) {
		Env env = engineConfig.getEnv();
		ZkConfig zkConfig = engineConfig.getZkConfig();

		String zkHost = zkConfig.getHost();
		int zkSessionTimeout = zkConfig.getSessionTimeout();

		String configPath = null;
		if (env.getEnv().equals("dev")) {
			configPath = DefaultConfig.ZK_DEV_CONF;
		}
		if (env.getEnv().equals("test")) {
			configPath = DefaultConfig.ZK_TEST_CONF;
		}
		if (env.getEnv().equals("staging")) {
			configPath = DefaultConfig.ZK_STAGING_CONF;
		}
		if (env.getEnv().equals("prod")) {
			configPath = DefaultConfig.ZK_PROD_CONF;
		}

		InputStream inputStream = null;
		Properties properties = new Properties();

		try {
			// 读取jar包中的文件流
			inputStream = ZKLockEngine.class.getResourceAsStream(configPath);
			properties.load(inputStream);

			if (StringUtils.isBlank(zkHost)) {
				host = properties.getProperty("host");
			} else {
				host = zkHost;
			}
			if (0 < zkSessionTimeout) {
				timeout = zkSessionTimeout;
			} else {
				timeout = Integer.valueOf(properties.getProperty("timeout"));
			}
			baseLockPath = properties.getProperty("baseLockPath");
		} catch (Exception e) {
			LOGGER.error("loadConfig fail", e);
		} finally {
			if (null != inputStream) {
				try {
					inputStream.close();
				} catch (IOException e) {
					//igonre
				}
			}
		}

	}
}
